/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: DoubleAndTripleBoltApp
 * Author:   An-Il
 * Date:     2018/11/9 9:19
 * Description: wordcount实例
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.local;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.Set;

/**
 * 使用storm实现wordcount实例
 *
 * @url https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/AnchoredWordCount.java#L16
 */
public class LocalWordCountApp {

    /**
     * 数据源产生的Spout
     */
    public static class RandomWordSpout extends BaseRichSpout {

        SpoutOutputCollector collector;
        Random random;

        /**
         * 初始化
         *
         * @param conf      配置参数
         * @param context   上下文
         * @param collector 数据发射器
         */
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector = collector;
            this.random = new Random();
        }

        /**
         * 此方法是个死循环
         */
        @Override
        public void nextTuple() {
            Utils.sleep(1000);
            String[] words = new String[]{"hadoop", "spark", "kafka", "flume", "storm", "hbase", "hive", "zookeeper"};
            String word = words[random.nextInt(words.length)];

            //将数据发送到下游
            this.collector.emit(new Values(word, 1));
        }

        /**
         * 申明输出字段
         *
         * @param declarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word", "count"));
        }
    }

    /**
     * 接受数据并处理
     */
    public static class WordCountBolt extends BaseRichBolt {

        OutputCollector collector;
        Map<String, Integer> countsMap;

        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
            countsMap = new HashMap<>();
        }

        @Override
        public void execute(Tuple tuple) {

            /**
             * 这里通过上面定义的Fields进行获取
             * 也可以通过下标获取：tuple.getInteger(0)
             */
            String word = tuple.getStringByField("word");
            Integer count = countsMap.get(word);
            if (count == null) {
                count = 0;
            }
            count++;

            //单词数汇总
            countsMap.put(word, count);

            //输出
            System.out.println("~~~~~~~~~~~~~~~~~~~~~~");
            Set<Map.Entry<String, Integer>> entrySet = countsMap.entrySet();
            for (Map.Entry<String, Integer> entry : entrySet) {
                System.out.println(entry);
            }


        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }
    }

    /**
     * 构建本地Topology
     *
     * @param args
     * @url http://storm.apache.org/releases/1.2.2/Local-mode.html
     * http://storm.apache.org/releases/1.2.2/Running-topologies-on-a-production-cluster.html
     */
    public static void main(String[] args) {

        /**
         * 定义拓扑,在生产群集上运行拓扑与在本地模式下运行类似
         * Topology需要指定相关的Spout和Bolt的执行顺序
         * shuffleGrouping("dataSourceSpout"):Tuples以一种随机分布方式在Bolt的任务中,每个Bolt都保证获得相同数量的Tuples
         */
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("RandomWordSpout", new RandomWordSpout());
        builder.setBolt("WordCountBolt", new WordCountBolt()).shuffleGrouping("RandomWordSpout");

        StormTopology topology = builder.createTopology();
        //本地模式模拟正在进行的Storm集群，只需使用LocalCluster类
        LocalCluster cluster = new LocalCluster();
        /**
         * 然后，您可以使用submitTopology该LocalCluster对象上的方法提交拓扑。就像StormSubmitter上的相应方法一样，
         * submitTopology采用名称，拓扑配置和拓扑对象。然后，您可以使用killTopology将拓扑名称作为参数的方法来终止拓扑。
         */
        cluster.submitTopology("LocalDoubleAndTripleBoltApp", new Config(), topology);


        //要关闭本地群集，请简单调用
        //cluster.shutdown();
    }

}